
#include <apr_general.h>
#include <apr_getopt.h>
#include <apr_strings.h>
#include <apr_signal.h>
#include "iface/APRInterface.h"
#include "iface/SSLInterface.h"
#include "iface/SQLiteInterface.h"

#include "aprtools/APRAutoMemoryPool.h"

#include "net/Broker5Client.h"
#include "utils/Hash.h"
#include "utils/StreamFromDirectory.h"
#include "utils/StreamToDirectory.h"
#include "utils/RunningWorkProcess.h"

#include "acquire_common.h"
#include "acquire_compute.h"
#include "default_hosts.h"

#include "acquire_common_private.h"

typedef struct compute_args
{
   char *brokername;
   char *hostname;
   char *hostport;
   char *directory;
   char *commkey;
   char *unencrypt;
   char *reencrypt;
   char *jobid;
   char *instanceid;
   int durseconds;
} ComputeArgs;

void processArguments(int argc, const char *argv[], ComputeArgs &args)
{
   static const apr_getopt_option_t opt_option[] = {
      { "host", 'n', TRUE, "hostname to connect to (broker host)" },
      { "port", 'p', TRUE, "port to connect to (broker port)" },
      { "directory", 'd', TRUE, "directory to run work in" },
      { "commkey", 'c', TRUE, "key to be used for communication" },
      { "unencrypt", 'u', TRUE, "key to be used for decryption of work" },
      { "reencrypt", 'r', TRUE, "key to be used for encryption of results" },
      { "jobid", 'j', TRUE, "job id assigned by system" },
      { "instanceid", 'i', TRUE, "job id assigned by system" },
      { "duration", 't', TRUE, "maximum duration of the compute" },
      { NULL, 0, 0, NULL } };
   apr_getopt_t *opt;
   int optch;
   const char *optarg = NULL;
   apr_status_t status;
   apr_pool_t *mp = NULL;
   status = apr_pool_create(&mp, NULL);
   CheckAPRError(status);
   args.hostname = apr_pstrdup(mp, DEFAULT_HOST);
   args.hostport = apr_pstrdup(mp, DEFAULT_PORT_STRING);
   args.directory = apr_pstrdup(mp, "/dev/null");
   args.commkey = apr_pstrdup(mp, "");
   args.unencrypt = apr_pstrdup(mp, "");
   args.reencrypt = apr_pstrdup(mp, "");
   args.jobid = apr_pstrdup(mp, "job");
   args.instanceid = apr_pstrdup(mp, "instance");
   args.durseconds = 3600 * 24;
   // 24 hours maximum time running - this should probably be changed (possible to infinite) !!!!
   apr_getopt_init(&opt, mp, argc, argv);
   while ((status = apr_getopt_long(opt, opt_option, &optch, &optarg)) == APR_SUCCESS)
   {
      switch (optch)
      {
         case 'n': args.hostname = apr_pstrdup(mp, optarg); break;
         case 'p': args.hostport = apr_pstrdup(mp, optarg); break;
         case 'd': args.directory = apr_pstrdup(mp, optarg); break;
         case 'c': args.commkey = apr_pstrdup(mp, optarg); break;
         case 'u': args.unencrypt = apr_pstrdup(mp, optarg); break;
         case 'r': args.reencrypt = apr_pstrdup(mp, optarg); break;
         case 'j': args.jobid = apr_pstrdup(mp, optarg); break;
         case 'i': args.instanceid = apr_pstrdup(mp, optarg); break;
         case 't': args.durseconds = atoi(apr_pstrdup(mp, optarg)); break;
         default:
         {
            const apr_getopt_option_t *options = opt_option;
            int count = 0;
            printf("Help for compute program:\n");
            printf("   progname");
            while (options[count].optch != '\0')
            {
               printf(" -%c", options[count].optch);
               if (options[count].has_arg)
                  printf(" <%s>", options[count].name);
               count++;
            }
            printf("\n\n");
            count = 0;
            while (options[count].optch != '\0')
            {
               printf("(-%c or --%s) - %s\n",
                  options[count].optch, options[count].name, options[count].description);
               count++;
            }
            return;
            break;
         }
      }
   }
   if (status != APR_EOF)
      CheckAPRError(status);
   // Never delete mp!
}

int main(int argc, const char *argv[])
{
   putARGV0ToGlobalPlace(argv[0]);
   int nargc = argc;
   const char *const *nargv = argv;
   acquireCommonInitLibraries();
   apr_signal_block(SIGPIPE);
   ComputeArgs args;
   processArguments(argc, argv, args);
   if (!(acquireComputeInitState(args.hostname, atoi(args.hostport), args.commkey,
      args.instanceid, "wwobble", args.commkey, NULL)))
   {
      printf("Error encountered while setting initial client state. Check log '%s' for possible reasons.\n", globalplace_logname);
      return 1;
   }
   acquireComputeUpdateUnencryptReencryptKeys(args.unencrypt, args.unencrypt, args.reencrypt, args.reencrypt, 1);
   acquireComputeDataStoreWorkInstanceInfo dswi_info;
   dswi_info = acquireComputeGetDataStoreForWorkInput();
   acquireCommonDataStoreInfo dsinfo;
   dsinfo = acquireCommonGetDataStoreInformation(dswi_info.datastoreid);
   int retcode = acquireCommonDownloadFromDataStore(args.directory, dsinfo.blockcount, dswi_info.datastoreid, NULL, args.unencrypt);
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   errprint("starting job");
   if (args.durseconds == 0)
   {
      errprint("FIX ME: number of seconds to live is zero (changing it to 24 hours!)");
      args.durseconds = 3600 * 24;
   }
   acquireComputeStartProcess(args.directory, dswi_info.localinstancenumber, dsinfo.instancecount);
   retcode = acquireComputeWatchForCancelOrComplete(args.durseconds);
   int64_t remote_space_remaining = -1;
   int blocks_to_upload = 0;
   int cancel = 0;
   const char *resultstoreid = acquireCommonCreateDataStore(args.directory,
      args.reencrypt, args.unencrypt,
      0, dsinfo.blocksize, &remote_space_remaining, &blocks_to_upload, &cancel);
   retcode = acquireCommonUploadToDataStore(resultstoreid, args.directory, dsinfo.blocksize,
      &blocks_to_upload, &cancel, args.reencrypt);
   if (retcode == 0)
   {
      acquireCommonSetPurposeFlagsOnDataStore(resultstoreid, 0, 1, NULL, 0, 0, NULL);
   }
   /*
   BrokerMessage *bmsg = new BrokerMessage(state.mp);
   bmsg->reset();
   bmsg->setRecipient(args.commkey);
   bmsg->setSender(args.commkey);
   bmsg->addTypeHeaderNOCRLF("START_DATA_STORE", args.instanceid);
   bmsg->addHeaderLineNOCRLF("Unencrypted-Block-Size: %d", dsinfo.blocksize);
   char *hexid = Hash::makeHexadecimalID(32);
   char *retval = NULL;
   bmsg->addHeaderLineNOCRLF("Data-Store-ID: %s", hexid);
   // Key pairs to create the new data store actor
   (state.keydb)->generateEncDecPairIntoDB(false, false, hexid, args.instanceid, AbstractKey::KEY_SYMMETRIC_AES);
   (state.keydb)->generateSgnChkPairIntoDB(false, false, false, hexid, args.instanceid, AbstractKey::KEY_ASYMMETRIC_PAIR_ECDSA);
   // Attach all the keys for the job, as we are creating a remote actor for this with keys
   bmsg->insertAllKeysToSend(0, hexid, NULL, state.keydb);
   int blocks = 0;
   EncryptionKey *enc_wrap_input_key = Key::getEncryptionKey(state.keydb, NULL, args.reencrypt);
   if (enc_wrap_input_key)
   {
      StreamFromDirectory s_from_dir(args.directory, pool.mp());
      EncryptWrapper enc_wrap_input(enc_wrap_input_key, NULL, true);
      int plain_size = dsinfo.blocksize;
      std::vector<char *> hashlist;
      std::vector<char *> cesizes;
      blocks = 0;
      char *block_data_buffer = (char *)malloc(sizeof(char)*dsinfo.blocksize);
      while (plain_size == dsinfo.blocksize)
      {
         char *thash = NULL;
         int compressed_encrypted_size = 0;
         char *this_block = s_from_dir.getNextBlockEncrypted(&enc_wrap_input, &plain_size, &compressed_encrypted_size, &thash);
         hashlist.push_back(thash);
         thash = NULL;
         cesizes.push_back(StringUtils::varprintf("%d", compressed_encrypted_size));
         free(this_block);
         blocks++;
      }
      char *all_hashes = StringUtils::flattenStringList(&hashlist, ",");
      char *all_sizes = StringUtils::flattenStringList(&cesizes, ",");
      // uses special destroyStringList to destroy the string entries.
      StringUtils::destroyStringList(hashlist);
      StringUtils::destroyStringList(cesizes);
      int hashes_payload_number = bmsg->addVerbatimPayloadString(all_hashes);
      int sizes_payload_number = bmsg->addVerbatimPayloadString(all_sizes);
      bmsg->addHeaderLineNOCRLF("Payload-Of-Hashes: %d", hashes_payload_number);
      bmsg->addHeaderLineNOCRLF("Payload-Of-Sizes: %d", sizes_payload_number);
      free(all_hashes);
      free(all_sizes);
      Broker5Client *brokerc = new Broker5Client(args.hostname, atoi(args.hostport), pool.mp());
      APRFuture<BrokerMessage *> *future_response = brokerc->exchangeMessageFuture(*bmsg, *(state.keydb));
      while (!(future_response->isFinished()));
      future_response->get();
      delete future_response;
      if (bmsg->getProcessee())
      {
         if (strcmp(bmsg->getType(), "DATA_STORE_DENIED") == 0)
         {
            if (strcmp(bmsg->getProcessee(), "REQUEST_INVALID") == 0)
               errprint("server reports invalid request");
            if (strcmp(bmsg->getProcessee(), "USER_HAS_NO_RESOURCES") == 0)
               errprint("user has no resources - must register an SSH login");
            if (strcmp(bmsg->getProcessee(), "ALREADY_EXISTS") == 0)
               errprint("data store already exists");
            if (strcmp(bmsg->getProcessee(), "OUT_OF_SPACE") == 0)
               errprint("user quota has run out - cannot allocate this new datastore");
         }
         //if (strcmp(bmsg->getType(), "DATA_STORE_CREATED") == 0)
         // {
         //   // need to get new quota size to update it locally
         //   retval = apr_pstrdup(state.mp, hexid);
         // }
      }
   }
   delete enc_wrap_input_key;
   int blocksdone = 0;
   int cancelflag = 0;
   int return_int = staticClientUploadToDataStore(args.hostname, atoi(args.hostport), state.keydb,
      hexid, args.directory, dsinfo.blocksize,
      &blocksdone, &cancelflag, args.reencrypt,
      pool.mp());
      */

   acquireCommonDestroyState();
   acquireCommonDestroy();
   return 0;
}
